package com.distributer.report.infrastructure.strom.bolt;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;
import com.distributer.report.domain.statistics.model.OrderStatistics;
import com.distributer.report.domain.statistics.valueobj.OrderValueObj;

/**
 * 订单总单量、总金额统计
 * 计算店铺的维度
        每个店铺的总销售额
   Redis的rowKey设计  distributer:order:shopId:price:date
        每个店铺的购买人数
   Redis的rowKey设计  distributer:order:shopId:user:date
       每个店铺的销售数量
   Redis的rowKey设计  distributer:order:shopId:num:date
   
       平台维度的数据统计
       平台总销售额度
   redisRowKey设计  distributer:order:total:price:date
      平台今天下单人数
   redisRowKey设计  distributer:order:total:user:date
       平台商品销售数量
   redisRowKey设计  distributer:order:total:num:date

 * @author xiaolin
 *
 */
public class OrderStatisticsBolt extends BaseRichBolt{

	private static final long serialVersionUID = 524924738993219573L;
	
	private static final Logger LOGGER = LoggerFactory.getLogger(OrderStatisticsBolt.class);
	
	private OutputCollector collector;
	private static OrderStatistics orderStatistics = new OrderStatistics();
	
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	@Override
	public void execute(Tuple input) {
		OrderValueObj orderValueObj = (OrderValueObj)input.getValueByField("orderValueObj");
		try {
			//LOGGER.info("OrderStatisticsBolt 准备处理订单信息： "+ JSONObject.toJSONString(orderValueObj));
			orderStatistics.statistics(orderValueObj);
			LOGGER.info(Thread.currentThread().getName()+ " OrderStatisticsBolt 产生的订单统计结果: "+JSONObject.toJSONString(orderStatistics));
			collector.ack(input);
			LOGGER.info(Thread.currentThread().getName()+" OrderStatisticsBolt ack => messageID:"+orderValueObj.getMessageId());
		} catch (Exception e) {
			collector.fail(input);
			LOGGER.info(Thread.currentThread().getName()+" OrderStatisticsBolt fail => messageID:"+orderValueObj.getMessageId(),e);
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		
	}

}
